/*
 * Copyright (C) 2011 TopCoder Inc., All Rights Reserved.
 */
package gov.nasa.pds.processors.impl;

import gov.nasa.pds.entities.Column;
import gov.nasa.pds.entities.DataFile;
import gov.nasa.pds.entities.DataSet;
import gov.nasa.pds.entities.Instrument;
import gov.nasa.pds.entities.InstrumentHost;
import gov.nasa.pds.entities.MetadataFile;
import gov.nasa.pds.entities.MetadataObject;
import gov.nasa.pds.entities.Mission;
import gov.nasa.pds.entities.Product;
import gov.nasa.pds.entities.Property;
import gov.nasa.pds.entities.Reference;
import gov.nasa.pds.entities.Table;
import gov.nasa.pds.entities.Target;
import gov.nasa.pds.entities.ValidationReport;
import gov.nasa.pds.entities.Volume;
import gov.nasa.pds.processors.DataSetProcessor;
import gov.nasa.pds.services.ConversionPersistence;
import gov.nasa.pds.services.DataFileCopier;
import gov.nasa.pds.services.DataFileReader;
import gov.nasa.pds.services.DataSetProcessingConfigurationException;
import gov.nasa.pds.services.DataSetProcessingException;
import gov.nasa.pds.services.MetadataFileReader;
import gov.nasa.pds.services.MetadataValidationManager;
import gov.nasa.pds.services.OffsetType;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.Set;

import org.springframework.beans.factory.InitializingBean;

import com.topcoder.commons.utils.LoggingWrapperUtility;
import com.topcoder.commons.utils.ValidationUtility;
import com.topcoder.util.log.Level;
import com.topcoder.util.log.Log;

/**
 * <p>
 * The <code>DataSetProcessorImpl</code> class implements <code>DataSetProcessor</code>
 * interface by providing implementation of <code>processDataSet()</code> method.
 * </p>
 *
 * <strong>Thread Safety:</strong> This class is mutable since it provides public setters for its properties. But it
 * doesn't change its state and is thread safe when the following conditions are met: this class is initialized by
 * Spring right after construction and its parameters are never changed after that, all entities passed to this class
 * are used by the caller in thread safe manner (accessed from a single thread only).
 *
 * @author KennyAlive
 * @version 1.0
 */
public class DataSetProcessorImpl implements DataSetProcessor, InitializingBean {
    /**
     * Constant for the class name of this class. Used for logging.
     */
    private static final String CLASS_NAME = DataSetProcessorImpl.class.getName();

    /**
     * The filename of the file with metadata for data set volume.
     */
    private static final String VOLUME_DESC_FILENAME = "voldesc.cat";

    /**
     * The name of the directory where catalog files are stored.
     */
    private static final String CATALOG_DIRECTORY_NAME = "catalog";

    /**
     * Constant for aareadme.txt file name
     */
    private static final String AAREADME_TXT_FILENAME = "aareadme.txt";

    /**
     * Constant for errata.txt file name
     */
    private static final String ERRATA_TXT_FILENAME = "errata.txt";

    /**
     * The relative path to the index.lbl file.
     */
    private static final String INDEX_LBL_REL_FILEPATH = "index/index.lbl";

    /**
     * The relative path to the index.tab file.
     */
    private static final String INDEX_TAB_REL_FILEPATH = "index/index.tab";

    /**
     * Suffix that indicates that the table offset is defined in bytes.
     */
    private static final String BYTES_SUFFIX = "<BYTES>";

    //--------- Begin of folders that need to be processed ---------
    /**
     * The document folder name
     */
    private static final String DOCUMENT_DIRECTORY = "document";
    /**
     * The calib folder name
     */
    private static final String CALIB_DIRECTORY = "calib";
    /**
     * The gazetter folder name
     */
    private static final String GAZETTER_DIRECTORY = "gazetter";
    /**
     * The software folder name
     */
    private static final String SOFTWARE_DIRECTORY = "software";

    //----------- Begin of object names -------------
    /**
     * The name of the volume metadata object - the toplevel object from the volume description file
     */
    private static final String VOLUME_OBJECT_NAME = "VOLUME";
    /**
     * The name of the catalog object that is defined within a volume object.
     */
    private static final String CATALOG_OBJECT_NAME = "CATALOG";
    /**
     * The name of the reference object.
     */
    private static final String REFERENCE_OBJECT_NAME = "REFERENCE";
    /**
     * The name of the instrument host object.
     */
    private static final String INSTRUMENT_HOST_OBJECT_NAME = "INSTRUMENT_HOST";
    /**
     * The name of the instrument object.
     */
    private static final String INSTRUMENT_OBJECT_NAME = "INSTRUMENT";
    /**
     * The name of the mission object.
     */
    private static final String MISSION_OBJECT_NAME = "MISSION";
    /**
     * The name of the target object.
     */
    private static final String TARGET_OBJECT_NAME = "TARGET";
    /**
     * The name of the data set object.
     */
    private static final String DATA_SET_OBJECT_NAME = "DATA_SET";
    /**
     * The name of the table object.
     */
    private static final String TABLE_OBJECT_NAME = "TABLE";
    /**
     * The name of the series object.
     */
    private static final String SERIES_OBJECT_NAME = "SERIES";
    /**
     * The name of the image object.
     */
    private static final String IMAGE_OBJECT_NAME = "IMAGE";
    /**
     * The name of the product object.
     */
    private static final String PRODUCT_OBJECT_NAME = "PRODUCT";


    //-------- Begin of catalog properties names --------
    /**
     * Reference property name of the catalog object.
     */
    private static final String CATALOG_REFERENCE_PROPERTY_NAME = "^REFERENCE_CATALOG";
    /**
     * Instrument host property name of the catalog object.
     */
    private static final String CATALOG_INSTRUMENT_HOST_PROPERTY_NAME = "^INSTRUMENT_HOST_CATALOG";
    /**
     * Instrument property name of the catalog object.
     */
    private static final String CATALOG_INSTRUMENT_PROPERTY_NAME = "^INSTRUMENT_CATALOG";
    /**
     * Mission property name of the catalog object.
     */
    private static final String CATALOG_MISSION_PROPERTY_NAME = "^MISSION_CATALOG";
    /**
     * Target property name of the catalog object.
     */
    private static final String CATALOG_TARGET_PROPERTY_NAME = "^TARGET_CATALOG";
    /**
     * Data set property name of the catalog object.
     */
    private static final String CATALOG_DATA_SET_PROPERTY_NAME = "^DATA_SET_CATALOG";


    /**
     * This set contains predefined properties names of the CATALOG metadata object.
     */
    private static final Set<String> CATALOG_PROPERTIES_NAMES;

    /**
     * Maps catalog property name to corresponding object name
     */
    private static final Map<String, String> CATALOG_PROPERTY_NAME_2_OBJECT_NAME;

    /**
     * Initializes <code>CATALOG_PROPERTIES_PATTERNS</code> with predefined patterns.
     * Initializes <code>CATALOG_PROPERTY_NAME_2_OBJECT_NAME</code>.
     */
    static {
        Set<String> set = new HashSet<String>();
        set.add(CATALOG_REFERENCE_PROPERTY_NAME);
        set.add(CATALOG_INSTRUMENT_HOST_PROPERTY_NAME);
        set.add(CATALOG_INSTRUMENT_PROPERTY_NAME);
        set.add(CATALOG_MISSION_PROPERTY_NAME);
        set.add(CATALOG_TARGET_PROPERTY_NAME);
        set.add(CATALOG_DATA_SET_PROPERTY_NAME);
        CATALOG_PROPERTIES_NAMES = Collections.unmodifiableSet(set);

        Map<String, String> map = new HashMap<String, String>();
        map.put(CATALOG_REFERENCE_PROPERTY_NAME, REFERENCE_OBJECT_NAME);
        map.put(CATALOG_INSTRUMENT_HOST_PROPERTY_NAME, INSTRUMENT_HOST_OBJECT_NAME);
        map.put(CATALOG_INSTRUMENT_PROPERTY_NAME, INSTRUMENT_OBJECT_NAME);
        map.put(CATALOG_MISSION_PROPERTY_NAME, MISSION_OBJECT_NAME);
        map.put(CATALOG_TARGET_PROPERTY_NAME, TARGET_OBJECT_NAME);
        map.put(CATALOG_DATA_SET_PROPERTY_NAME, DATA_SET_OBJECT_NAME);
        CATALOG_PROPERTY_NAME_2_OBJECT_NAME = Collections.unmodifiableMap(map);
    }

    /**
     * <p>
     * This enumeration provides product pointers types.
     * </p>
     *
     * NOTE: the implementation uses the following convention: enumerator names match the corresponding
     * pointer names without leading ^ symbol and optional prefix.
     *
     * <strong>Thread Safety:</strong> This enum is immutable and thread safe.
     *
     * @author TCSASSEMBLER
     * @version 1.0
     */
    private enum ProductPointer {
        TABLE,
        SERIES,
        LABEL,
        TEXT,
        HEADER,
        IMAGE
    }

    /**
     * <p>
     * The <code>Log</code> instance used for logging.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private Log logger;

    /**
     * <p>
     * The path to the root directory of the data set.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private String sourceRootDirectoryName;

    /**
     * <p>
     * The destination directory for files copying.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private String destinationRootDirectoryName;

    /**
     * <p>
     * The <code>MetadataFileReader</code> instance.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private MetadataFileReader metadataFileReader;

    /**
     * <p>
     * The <code>MetadataValidationManager</code> instance.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private MetadataValidationManager metadataValidationManager;

    /**
     * <p>
     * The <code>DataFileCopier</code> instance.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private DataFileCopier dataFileCopier;

    /**
     * <p>
     * The <code>DataFileReader</code> instance.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private DataFileReader dataFileReader;

    /**
     * <p>
     * The <code>ConversionPersistence</code> instance.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private ConversionPersistence conversionPersistence;

    /**
     * <p>
     * The file types that should have their content read. Anything else should be copied as-is to a
     * destination folder.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private Set<String> asciiFileTypes;

    /**
     * <p>
     * The file types that should be ignored when processing the software folder.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private Set<String> ignoredSoftwareFileTypes;

    /**
     * <p>
     * The line separator symbols. It is an optional property. If it is not specified then line.separator system
     * property is used.
     * </p>
     *
     * <p>
     * It is initialized with Spring setter dependency injection. Cannot be <code>null</code> after initialization,
     * assuming that property is initialized via Spring setter-based dependency injection and is never changed
     * after that. Has a setter.
     * </p>
     */
    private String lineSeparator;

    /**
     * Creates an instance of <code>DataSetProcessorImpl</code>
     */
    public DataSetProcessorImpl() {
        // Empty
    }

    /**
     * Sets the logger.
     *
     * @param logger
     *              the logger to set
     */
    public void setLogger(Log logger) {
        this.logger = logger;
    }

    /**
     * Sets the source root directory name.
     *
     * @param sourceRootDirectoryName
     *              the source root directory name to set
     */
    public void setSourceRootDirectoryName(String sourceRootDirectoryName) {
        this.sourceRootDirectoryName = sourceRootDirectoryName;
    }

    /**
     * Sets the destination directory for files copying.
     *
     * @param destinationRootDirectoryName
     *              the destination directory to set
     */
    public void setDestinationRootDirectoryName(String destinationRootDirectoryName) {
        this.destinationRootDirectoryName = destinationRootDirectoryName;
    }

    /**
     * Sets the metadata file reader.
     *
     * @param metadataFileReader
     *              the metadata file reader to set
     */
    public void setMetadataFileReader(MetadataFileReader metadataFileReader) {
        this.metadataFileReader = metadataFileReader;
    }

    /**
     * Sets the metadata validation manager.
     *
     * @param metadataValidationManager
     *              the metadata validation manager to set
     */
    public void setMetadataValidationManager(MetadataValidationManager metadataValidationManager) {
        this.metadataValidationManager = metadataValidationManager;
    }

    /**
     * Sets the data file copier.
     *
     * @param dataFileCopier
     *              the data file copier to set
     */
    public void setDataFileCopier(DataFileCopier dataFileCopier) {
        this.dataFileCopier = dataFileCopier;
    }

    /**
     * Sets the data file reader.
     *
     * @param dataFileReader
     *              the data file reader to set
     */
    public void setDataFileReader(DataFileReader dataFileReader) {
        this.dataFileReader = dataFileReader;
    }

    /**
     * Sets the conversion persistence instance.
     *
     * @param conversionPersistence
     *              the conversion persistence instance to set
     */
    public void setConversionPersistence(ConversionPersistence conversionPersistence) {
        this.conversionPersistence = conversionPersistence;
    }

    /**
     * Sets the ASCII file types.
     *
     * @param asciiFileTypes
     *              the ASCII files types that should have their content read
     */
    public void setAsciiFileTypes(Set<String> asciiFileTypes) {
        this.asciiFileTypes = asciiFileTypes;
    }

    /**
     * Sets the ignored software file types.
     *
     * @param ignoredSoftwareFileTypes
     *              the files types that should be ignored when processing the software folder
     */
    public void setIgnoredSoftwareFileTypes(Set<String> ignoredSoftwareFileTypes) {
        this.ignoredSoftwareFileTypes = ignoredSoftwareFileTypes;
    }

    /**
     * Sets the line separator that is used when creating a string that represents the content of the text file.
     *
     * @param lineSeparator
     *              the line separator to set
     */
    public void setLineSeparator(String lineSeparator) {
        this.lineSeparator = lineSeparator;
    }

    /**
     * Checks whether this class was initialized by Spring properly.
     *
     * Required parameters:
     * <ul>
     * <li>logger</li>
     * <li>sourceRootDirectoryName</li>
     * <li>destinationRootDirectoryName</li>
     * <li>metadataFileReader</li>
     * <li>metadataValidationManager</li>
     * <li>dataFileCopier</li>
     * <li>dataFileReader</li>
     * <li>conversionPersistence</li>
     * <li>asciiFileTypes</li>
     * </ul>
     *
     * Optional parameter: ignoredSoftwareFileTypes
     *
     * @throws DataSetProcessingConfigurationException
     *            if the class was not initialized properly
     */
    @Override
    public void afterPropertiesSet() {
        ValidationUtility.checkNotNull(logger, "logger",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(sourceRootDirectoryName, "sourceRootDirectoryName",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(destinationRootDirectoryName, "destinationRootDirectoryName",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(metadataFileReader, "metadataFileReader",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(metadataValidationManager, "metadataValidationManager",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(dataFileCopier, "dataFileCopier",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(dataFileReader, "dataFileReader",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(conversionPersistence, "conversionPersistence",
                DataSetProcessingConfigurationException.class);
        ValidationUtility.checkNotNull(asciiFileTypes, "asciiFileTypes",
                DataSetProcessingConfigurationException.class);
    }

    /**
     * Processes the configured data set. The data set that should be processed is specified via
     * <code>sourceRootDirectoryName</code> injected parameter which represents a path to the root
     * directory of the data set of interest.
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    @Override
    public void processDataSet() throws DataSetProcessingException {
        String signature = CLASS_NAME + ".processDataSet()";
        LoggingWrapperUtility.logEntrance(logger, signature, null, null);

        metadataValidationManager.load();
        
        // 1. Process volume metadata (VOLDESC.CAT)
        long dataSetId = processVolumeDescriptor();

        // 2. Process root directory textual files (AAREADME.TXT, ERRATA.TXT)
        processRootDirectoryTextFiles(dataSetId);

        // 3. Process index and products
        processIndexAndProducts(dataSetId);

        // 4. Process other files
        processDirectory(DOCUMENT_DIRECTORY, dataSetId, false);
        processDirectory(CALIB_DIRECTORY,    dataSetId, false);
        processDirectory(GAZETTER_DIRECTORY, dataSetId, false);
        processDirectory(SOFTWARE_DIRECTORY, dataSetId, true);

        LoggingWrapperUtility.logExit(logger, signature, null);
    }

    /**
     * Process volume descriptor file (VOLDESC.CAT).
     *
     * @return the identifier of the data set object
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private long processVolumeDescriptor() throws DataSetProcessingException {
        File volumeDescFile = new File(sourceRootDirectoryName, VOLUME_DESC_FILENAME);
        MetadataFile metadataFile = metadataFileReader.readMetadataInfo(volumeDescFile.getPath());
        MetadataObject volumeMetadata = getRequiredMetadataObject(metadataFile, VOLUME_OBJECT_NAME);
        validateAndCorrectMetadata(Arrays.asList(volumeMetadata));

        Volume volume = new Volume(VOLUME_OBJECT_NAME);
        volume.fromMetadata(volumeMetadata);
        conversionPersistence.insertVolume(volume);

        return processVolumeCatalog(volumeMetadata);
    }

    /**
     * Processes CATALOG MetadataObject defined by the given VOLUME MetadataObject.
     *
     * @param volumeMetadata
     *              the volume metadata object that defines the catalog object
     *
     * @return the identifier of the data set object
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private long processVolumeCatalog(MetadataObject volumeMetadata)
        throws DataSetProcessingException {
        final String signature = CLASS_NAME + ".processVolumeCatalog(MetadataObject volumeMetadata)";

        long dataSetId = -1L;

        MetadataObject catalogMetadata = getRequiredMetadataObject(volumeMetadata, CATALOG_OBJECT_NAME);

        for (Property property : catalogMetadata.getProperties()) {
            if (!CATALOG_PROPERTIES_NAMES.contains(property.getName())) {
                String message = String.format("Unknown property (%s) found in CATALOG object", property.getName());
                throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message));
            }
            String propertyName = property.getName();
            if (property.getValues().get(0).equals("N/A")) {
                logger.log(Level.WARN, "Property {0} of CATALOG object has invalid value N/A", propertyName);
                continue;
            }
            File catalogDirectory = new File(sourceRootDirectoryName, CATALOG_DIRECTORY_NAME);
            String filename = property.getValues().get(0).toLowerCase();
            String filepath = (new File(catalogDirectory, filename)).getPath();

            MetadataFile metadataFile = metadataFileReader.readMetadataInfo(filepath);

            String metadataObjectName = CATALOG_PROPERTY_NAME_2_OBJECT_NAME.get(propertyName);
            List<MetadataObject> metadataObjects = getMetadataObjects(metadataFile, metadataObjectName);
            validateAndCorrectMetadata(metadataObjects);

            if (propertyName.equals(CATALOG_REFERENCE_PROPERTY_NAME)) {
                for (MetadataObject metadataObject : metadataObjects) {
                    Reference reference = new Reference();
                    reference.fromMetadata(metadataObject);
                    conversionPersistence.insertReference(reference);
                }
            } else if (propertyName.equals(CATALOG_INSTRUMENT_HOST_PROPERTY_NAME)) {
                for (MetadataObject metadataObject : metadataObjects) {
                    InstrumentHost instrumentHost = new InstrumentHost(INSTRUMENT_HOST_OBJECT_NAME);
                    instrumentHost.fromMetadata(metadataObject);
                    conversionPersistence.insertInstrumentHost(instrumentHost);
                }
            } else if (propertyName.equals(CATALOG_INSTRUMENT_PROPERTY_NAME)) {
                for (MetadataObject metadataObject : metadataObjects) {
                    Instrument instrument = new Instrument();
                    instrument.fromMetadata(metadataObject);
                    conversionPersistence.insertInstrument(instrument);
                }
            } else if (propertyName.equals(CATALOG_MISSION_PROPERTY_NAME)) {
                for (MetadataObject metadataObject : metadataObjects) {
                    Mission mission = new Mission(MISSION_OBJECT_NAME);
                    mission.fromMetadata(metadataObject);
                    conversionPersistence.insertMission(mission);
                }
            } else if (propertyName.equals(CATALOG_TARGET_PROPERTY_NAME)) {
                for (MetadataObject metadataObject : metadataObjects) {
                    Target target = new Target(TARGET_OBJECT_NAME);
                    target.fromMetadata(metadataObject);
                    conversionPersistence.insertTarget(target);
                }
            } else if (propertyName.equals(CATALOG_DATA_SET_PROPERTY_NAME)) {
                if (metadataObjects.size() > 1) {
                    throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                            "More than one data set objects are defined"));
                } else if (metadataObjects.isEmpty()) {
                    throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                            "The data set object is not defined"));
                }
                DataSet dataSet = new DataSet(DATA_SET_OBJECT_NAME);
                dataSet.fromMetadata(metadataObjects.get(0));
                dataSetId = conversionPersistence.insertDataSet(dataSet);
            } else {
                throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                        "Unexpected property name"));
            }
        }
        if (dataSetId == -1L) {
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                    "The data set object is not defined"));
        }
        return dataSetId;
    }

    /**
     * Processes text files in the root directory of the data set.
     *
     * @param dataSetId
     *              the data set id
     *
     * @throws DataSetProcessingException
     *              if there is any persistence error while processing data set
     */
    private void processRootDirectoryTextFiles(long dataSetId) throws DataSetProcessingException {
        // process aareadme.txt: if it doesn't exist log a warning
        File file = new File(sourceRootDirectoryName, AAREADME_TXT_FILENAME);
        if (file.exists()) {
            DataFile dataFile = createDataFileFromFile(file, file.getName());
            conversionPersistence.insertDataSetDocument(dataSetId, dataFile);
        } else {
            logger.log(Level.WARN, "Readme file {0} is not found in {1} directory",
                    AAREADME_TXT_FILENAME, sourceRootDirectoryName);
        }
        // process errata.txt
        file = new File(sourceRootDirectoryName, ERRATA_TXT_FILENAME);
        if (file.exists()) {
            DataFile dataFile = createDataFileFromFile(file, file.getName());
            conversionPersistence.insertDataSetDocument(dataSetId, dataFile);
        }
    }

    /**
     * Processes index and products.
     *
     * @param dataSetId
     *              the data set id
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processIndexAndProducts(long dataSetId) throws DataSetProcessingException {
        File file = new File(sourceRootDirectoryName, INDEX_LBL_REL_FILEPATH);
        MetadataFile metadataFile = metadataFileReader.readMetadataInfo(file.getPath());

        MetadataObject metadataObject = getRequiredMetadataObject(metadataFile, TABLE_OBJECT_NAME);
        Table indexTable = new Table();
        indexTable.fromMetadata(metadataObject);

        file = new File(sourceRootDirectoryName, INDEX_TAB_REL_FILEPATH);
        List<String> associatedLabelFilenames = getAssociatedLabelFilenames(file.getPath(), indexTable);

        for (String associatedLabelFilename : associatedLabelFilenames) {
            file = new File(sourceRootDirectoryName, associatedLabelFilename);
            MetadataFile associatedMetadataFile = metadataFileReader.readMetadataInfo(file.getPath());

            Product product = new Product(PRODUCT_OBJECT_NAME);
            product.fromMetadata(associatedMetadataFile);
            long productId = conversionPersistence.insertProduct(dataSetId, product);

            // simply use the object's field as the storage for id in order to, not to pass this
            // id as an additional argument in processProduct() and other methods. It's safe since
            // another services don't use this product object anymore.
            product.setId(productId);

            processProduct(product, new File(associatedLabelFilename).getParentFile());
        }
    }

    /**
     * Processes the given product.
     *
     * @param product
     *              the product to process
     * @param dataDirectory
     *              the data directory relative to the <code>sourceRootDirectoryName</code> directory
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processProduct(Product product, File dataDirectory) throws DataSetProcessingException {
        for (Property property : product.getOtherProperties()) {
            String propertyName = property.getName();
            if (!propertyName.startsWith("^")) {
                continue;
            }
            // get property name without prefix and without leading ^
            int underscoreIndex = propertyName.lastIndexOf('_');
            if (underscoreIndex != -1) {
                propertyName = propertyName.substring(underscoreIndex + 1);
            } else {
                propertyName = propertyName.substring(1); // skip leading ^ symbol
            }
            // if propertyName matches one of the ProductPointer enumerators then we have found a pointer
            for (ProductPointer pointer : ProductPointer.values()) {
                if (propertyName.equals(pointer.name())) {
                    // master data file relative to the sourceRootDirectoryName directory
                    File masterDataFile = new File(dataDirectory, property.getValues().get(0));

                    // process pointer depending on its type
                    if (pointer == ProductPointer.TABLE
                            || pointer == ProductPointer.SERIES) {
                        masterDataFile = new File(sourceRootDirectoryName, masterDataFile.getPath());
                        processProductTable(product, pointer, masterDataFile.getPath(), property);
                    } else if (pointer == ProductPointer.LABEL
                            || pointer == ProductPointer.TEXT
                            || pointer == ProductPointer.HEADER) {
                        masterDataFile = new File(sourceRootDirectoryName, masterDataFile.getPath());
                        processProductTextDocument(product, pointer, masterDataFile.getPath());
                    } else if (pointer == ProductPointer.IMAGE) {
                        processProductImage(product, masterDataFile.getPath());
                    }
                }
            }
        }
    }

    /**
     * Processes a table associated with the given product.
     *
     * @param product
     *              the product that declares a table
     * @param pointer
     *              product pointer enumerator that defines table type: TABLE or SERIES
     * @param masterDataFilepath
     *              the filepath of the table data file
     * @param property
     *              the property that represents a pointer to this table/series product
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processProductTable(Product product, ProductPointer pointer, String masterDataFilepath,
         Property property) throws DataSetProcessingException {
        final String signature = CLASS_NAME + ".processProductTable(Product product, ProductPointer pointer,"
                                            + "String masterDataFilepath, Property property)";

        MetadataObject tableMetadata = null;

        // locate 'TABLE' or 'SERIES' object depending on the pointer type
        for (MetadataObject object : product.getOtherChildren()) {
            if ((pointer == ProductPointer.TABLE && object.getName().equals(TABLE_OBJECT_NAME))
            ||  (pointer == ProductPointer.SERIES && object.getName().equals(SERIES_OBJECT_NAME))) {
                tableMetadata = object;
                break;
            }
        }
        if (tableMetadata == null) {
            String message = String.format("Failed to find metadata object for data file '%s'", masterDataFilepath);
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message));
        }

        // check if offset is specified for this table
        Integer offset = null;
        OffsetType offsetType = null;
        if (property.getValues().size() > 1) {
            String offsetString = property.getValues().get(1);
            if (offsetString.toUpperCase().endsWith(BYTES_SUFFIX)) {
                offsetType = OffsetType.BYTES;
                int endPos = offsetString.length() - BYTES_SUFFIX.length();
                offsetString = offsetString.substring(0, endPos).trim();
            } else {
                offsetType = OffsetType.RECORD;
            }
            try {
                offset = Integer.valueOf(offsetString);
            } catch (NumberFormatException e) {
                String message = String.format("Invalid format of table offset: '%s'",
                                               property.getValues().get(1));
                throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                        message, e));
            }
        }

        // create new table with found metadata
        String tableName = generateNewObjectName(tableMetadata.getName(), masterDataFilepath);
        Table table = new Table();
        table.setName(tableName);
        table.fromMetadata(tableMetadata);
        // fill in table with data and store it in the persistence
        conversionPersistence.createTableStructure(table);
        dataFileReader.readData(product, masterDataFilepath, table, offset, offsetType);
        conversionPersistence.insertDataIntoTable(table);
        // Notice, that we set product.id earlier in order not to pass it as the method argument
        conversionPersistence.associateTableToProduct(product.getId(), tableName);
    }

    /**
     * Processes text information associated with the given product.
     *
     * @param product
     *              the product that declares a reference to the text file
     * @param pointer
     *               product pointer enumerator that defines object type: LABEL, TEXT or HEADER
     * @param masterDataFilepath
     *              the filepath of the text file
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processProductTextDocument(Product product, ProductPointer pointer, String masterDataFilepath)
        throws DataSetProcessingException {
        // Since for LABEL/TEXT/HEADER pointers there is no corresponding metadata object definition,
        // we use pointer name (without ^ and prefix) as the object name.
        String objectName = pointer.name();

        String dataFileName = generateNewObjectName(objectName, masterDataFilepath);
        File file = new File(masterDataFilepath);
        DataFile dataFile = createDataFileFromFile(file, dataFileName);
        // Notice, that we set product.id earlier in order not to pass it as the method argument
        conversionPersistence.insertProductDocument(product.getId(), dataFile);
    }

    /**
     * Processes image associated with the given product.
     *
     * @param product
     *              the product that declares a reference to the image file
     * @param relativeMasterDataFilepath
     *              the filepath of the image file relative to the <code>sourceRootDirectoryName</code> directory
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processProductImage(Product product, String relativeMasterDataFilepath)
        throws DataSetProcessingException {
        String dataFileName = generateNewObjectName(IMAGE_OBJECT_NAME, relativeMasterDataFilepath);
        DataFile dataFile = new DataFile(dataFileName);
        String destination = generateNewImagePath(relativeMasterDataFilepath);

        // make a copy of the original file and store it in the destinationRootDirectoryName directory
        dataFile.setPath(new File(sourceRootDirectoryName, relativeMasterDataFilepath).getPath());
        dataFileCopier.copyDataFile(dataFile, destination);

        // notice, that we set product.id earlier in order not to pass it as the method argument
        dataFile.setPath(destination);
        conversionPersistence.insertProductDocument(product.getId(), dataFile);
    }

    /**
     * Retrieves associated label files from a given index.tab file.
     *
     * @param indexTabFilename
     *              the full filename of the index.tab that should be processed
     * @param indexTable
     *              the index table that defines control parameters
     *
     * @return  the list of filenames of the associated label files
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing error while processing data set
     */
    private List<String> getAssociatedLabelFilenames(String indexTabFilename, Table indexTable)
        throws DataSetProcessingException {
        final String signature = CLASS_NAME
                + ".getAssociatedLabelFilenames(String indexTabFilename, Table indexTable)";

        final String LABEL_FILENAME_COLUMN_NAME = "FILE_SPECIFICATION_NAME";

        // determine column that contains associated label filenames
        int columnIndex = 0;
        for (Column column : indexTable.getColumns()) {
            if (column.getName().equals(LABEL_FILENAME_COLUMN_NAME)) {
                break;
            }
            columnIndex++;
        }
        if (columnIndex == indexTable.getColumns().size()) {
            String message = "Failed to find column in index table: " + LABEL_FILENAME_COLUMN_NAME;
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message));
        }

        // open index.tab and read it line by line
        List<String> filenames = new ArrayList<String>();
        try {
            BufferedReader reader = new BufferedReader(new FileReader(indexTabFilename));
            int rowCount = 0;
            String line = reader.readLine();
            while (line != null) {
                Scanner scanner = new Scanner(line);
                scanner.useDelimiter("(\\s|,|\")+");
                String filename = null;
                for (int i = 0; i <= columnIndex; i++) {
                    if (!scanner.hasNext()) {
                        String message = String.format("Invalid '%s', row %d is malformed",
                                                       indexTabFilename, rowCount + 1);
                        throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                                message));
                    }
                    filename = scanner.next();
                }
                filenames.add(filename);
                line = reader.readLine();
                rowCount++;
            }
            reader.close();
            if (rowCount != indexTable.getRowCount()) {
                String message = String.format("Rows count mismatch: index.lbl (%d), index.tab (%d)",
                                               indexTable.getRowCount(), rowCount);
                throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message));
            }
        } catch (IOException e) {
            String message = String.format("Failed to read index.tab file: '%s'", indexTabFilename);
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message, e));
        }
        return filenames;
    }

    /**
     * Processes the files from the given folder.
     *
     * @param directoryName
     *              specifies the name of the directory to process
     * @param dataSetId
     *              the data set id
     * @param softwareDirectory
     *              defines whether the given directory is a software directory
     *
     * @throws DataSetProcessingException
     *              if there is any IO/parsing/persistence error while processing data set
     */
    private void processDirectory(String directoryName, long dataSetId, boolean softwareDirectory)
        throws DataSetProcessingException {
        final String signature = CLASS_NAME +
                ".processDirectory(String directoryName, long dataSetId, boolean softwareDirectory)";

        // check source directory
        File sourceDirectory = new File(sourceRootDirectoryName, directoryName);
        if (!sourceDirectory.exists()) {
            return;
        }
        // check destination directory
        File destinationDirectory = new File(destinationRootDirectoryName, directoryName);
        if (!destinationDirectory.exists() && !destinationDirectory.mkdir()) {
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                    "Failed to create directory " + destinationDirectory.getPath()));
        }

        // get all not ".lbl" files from the source directory and all sub-directories
        List<File> files = new ArrayList<File>();
        listFiles(sourceDirectory, files);
        for (File file : files) {
            String extension = getFileExtension(file);
            // for software folder ignore file types defined by ignoredSoftwareFileTypes
            if (softwareDirectory && ignoredSoftwareFileTypes.contains(extension)) {
                continue;
            }
            if (asciiFileTypes.contains(extension)) {
                DataFile dataFile = createDataFileFromFile(file, file.getName());
                conversionPersistence.insertDataSetDocument(dataSetId, dataFile);
            } else {
                DataFile dataFile = new DataFile(file.getName());
                File destination = new File(destinationDirectory, trimPath(file, directoryName).getPath());

                dataFile.setPath(file.getPath());
                dataFileCopier.copyDataFile(dataFile, destination.getPath());

                dataFile.setPath(destination.getPath());
                conversionPersistence.insertDataSetDocument(dataSetId, dataFile);
            }
        }
    }

    /**
     * Gets all non ".lbl" files from the given directory and all its sub-directories.
     *
     * @param directory
     *              the directory to scan for files and sub-directories
     * @param result
     *              the resulted list where the found files are stored
     */
    private void listFiles(File directory, List<File> result) {
        File[] directoryFiles = directory.listFiles();
        // at first get the files from the current directory
        for (File file : directoryFiles) {
            if (file.isFile() && !file.getName().endsWith(".lbl")) {
                result.add(file);
            }
        }
        // add files from sub-directories only after the files from the current directory
        // were added in order to not intermix files from different directories
        for (File file : directoryFiles) {
            if (file.isDirectory()) {
                listFiles(file, result);
            }
        }
    }

    /**
     * Retrieves the first child metadata object with a given name from the parent object.
     *
     * @param parent
     *              the parent metadata object that holds the object of interest
     * @param childObjectName
     *              the name of the child object we are looking for
     *
     * @return the first child metadata object with a given name
     *
     * @throws DataSetProcessingException
     *              if the child object is not found
     */
    private MetadataObject getRequiredMetadataObject(MetadataObject parent, String childObjectName)
        throws DataSetProcessingException {
        final String signature = CLASS_NAME +
                ".getRequiredMetadataObject(MetadataObject parent, String childObjectName)";

        for (MetadataObject child : parent.getChildren()) {
            if (child.getName().equals(childObjectName)) {
                return child;
            }
        }
        String message = String.format("Failed to find a child object (%s) in the parent (%s) object",
                                       childObjectName, parent.getName());
        throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message));
    }

    /**
     * Retrieves all metadata objects with a given name from the parent object.
     *
     * @param parent
     *              the parent metadata object that holds the objects of interest
     * @param childObjectName
     *              the name of the child objects we are looking for
     *
     * @return the child metadata objects with a given name
     */
    private List<MetadataObject> getMetadataObjects(MetadataObject parent, String childObjectName) {
        List<MetadataObject> childObjects = new ArrayList<MetadataObject>();
        for (MetadataObject child : parent.getChildren()) {
            if (child.getName().equals(childObjectName)) {
                childObjects.add(child);
            }
        }
        return childObjects;
    }

    /**
     * Validates specified metadata objects and if necessary corrects them.
     *
     * @param metadataObjects
     *              the metadata objects to validate and correct
     *
     * @throws DataSetProcessingException
     *              if there is an error while doing validation
     */
    private void validateAndCorrectMetadata(List<MetadataObject> metadataObjects) throws DataSetProcessingException {
        for (MetadataObject metadataObject : metadataObjects) {
            ValidationReport report = metadataValidationManager.validateMetadata(metadataObject);
            if (!report.isValid()) {
                metadataValidationManager.correctMetadata(metadataObject, report);
            }
        }
    }

    /**
     * Creates a data file object and initializes it with the contents of the given file.
     *
     * @param file
     *              the file to read from
     * @param dataFileName
     *              specifies the name of the data file object
     *
     * @return the data file object with content initialized from the given file
     *
     * @throws DataSetProcessingException
     *              if there is an error while reading the file
     */
    private DataFile createDataFileFromFile(File file, String dataFileName) throws DataSetProcessingException {
        final String signature = CLASS_NAME + ".createDataFileFromFile(File file, String dataFileName)";

        // if the lineSeparator optional spring-injected property is not defined then
        // use line.separator system property
        if (lineSeparator == null) {
            lineSeparator = System.getProperty("line.separator");
        }

        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader(file));
            StringBuilder sb = new StringBuilder();

            String line = reader.readLine();
            while (line != null) {
                sb.append(line).append(lineSeparator);
                line = reader.readLine();
            }

            DataFile dataFile = new DataFile(dataFileName);
            dataFile.setContent(sb.toString());
            return dataFile;
        } catch (IOException e) {
            String message = String.format("Failed to read '%s' file", file.getPath());
            throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(message, e));
        } finally {
            try {
                if (reader != null) {
                    reader.close();
                }
            } catch (IOException e) {
                String message = String.format("Failed to close stream associated with '%s' file", file.getPath());
                throw LoggingWrapperUtility.logException(logger, signature, new DataSetProcessingException(
                        message, e));
            }
        }
    }

    /**
     * Returns the extension of the given file without leading dot symbol.
     *
     * @param file
     *              the file to extract extension from
     *
     * @return the extension of the given file or an empty string if the file has no extension
     */
    private String getFileExtension(File file) {
        String filename = file.getName();
        int dotIndex = filename.lastIndexOf('.');
        if (dotIndex == -1) {
            return "";
        }
        return filename.substring(dotIndex + 1);
    }

    /**
     * Extracts the part of the path that follows after the specified directory name.
     *
     * @param file
     *              the file that defines a path
     * @param directoryName
     *              the directory that represents reference point for path trimming
     *
     * @return the part of the original path that goes after the specified directory
     *
     */
    private File trimPath(File file, String directoryName) {
        File pathAfterDirectory = new File("");
        while (!file.getName().equals(directoryName)) {
            pathAfterDirectory = new File(file.getName(), pathAfterDirectory.getPath());
            file = file.getParentFile();
        }
        return pathAfterDirectory;
    }

    /**
     * Generates new object name using the following pattern: <data_filename>_<object_name>
     *
     * @param objectName
     *              the object name
     * @param masterDataFilepath
     *              the filepath of the master data file
     *
     * @return the new object name
     */
    private String generateNewObjectName(String objectName, String masterDataFilepath) {
        // get filename without extension
        String filename = (new File(masterDataFilepath)).getName();
        int dotIndex = filename.lastIndexOf('.');
        if (dotIndex != -1) {
            filename = filename.substring(0, dotIndex);
        }
        // create new name
        return filename + "_" + objectName;
    }

    /**
     * Generates new filepath for the image with a given filepath using the following
     * pattern: <destination_directory_path>/<original_filename>_<IMAGE><original_extension_with_dot>
     *
     * @param relativeMasterDataFilepath
     *              the filepath of the image master data file relative to the <code>sourceRootDirectoryName</code>
     *              directory
     *
     * @return the new image filepath
     */
    private String generateNewImagePath(String relativeMasterDataFilepath) {
        File file = new File(relativeMasterDataFilepath);
        String filename = file.getName();
        int dotIndex = filename.lastIndexOf('.');
        String extension = "";
        if (dotIndex != -1) {
            extension = filename.substring(dotIndex);
            filename = filename.substring(0, dotIndex);
        }
        String newFilename = filename + "_" + IMAGE_OBJECT_NAME + extension;

        File newFile = new File(destinationRootDirectoryName, file.getParentFile().getPath());
        newFile = new File(newFile, newFilename);
        return newFile.getPath();
     }
}
